#include <algorithm>
#include <chrono>
#include <cstddef>
#include <cstdint>
#include <deque>
#include <functional>
#include <iomanip>
#include <iostream>
#include <mutex>
#include <string>
#include <vector>
#if defined(__APPLE__) || defined(__linux__)
#include <cstdlib>
#include <unistd.h>
#endif

#include "asycipc/api.hpp"
#include "asycipc/frame.hpp"
#include "asycipc/named_pipe.hpp"

namespace {

using namespace asycipc;

std::vector<uint8_t> ToBytes(const std::string& text) {
  return std::vector<uint8_t>(text.begin(), text.end());
}

std::string ToString(const std::vector<uint8_t>& data) {
  return std::string(data.begin(), data.end());
}

std::string DefaultEndpoint() {
#if defined(_WIN32)
  return R"(\\.\pipe\asycipc_demo)";
#elif defined(__APPLE__)
  if (const char* override_dir = std::getenv("ASYIPC_RUNTIME_DIR")) {
    if (*override_dir) {
      return std::string(override_dir) + "/demo";
    }
  }
  if (geteuid() == 0) {
    return "/var/run/asycipc/demo";
  }
  if (const char* xdg = std::getenv("XDG_RUNTIME_DIR")) {
    if (*xdg) {
      return std::string(xdg) + "/asycipc/demo";
    }
  }
  if (const char* tmp = std::getenv("TMPDIR")) {
    if (*tmp) {
      return std::string(tmp) + "/asycipc/demo";
    }
  }
  return std::string("/tmp/asycipc/demo");
#else
  if (const char* override_dir = std::getenv("ASYIPC_RUNTIME_DIR")) {
    if (*override_dir) {
      return std::string(override_dir) + "/demo";
    }
  }
  if (geteuid() == 0) {
    return "/run/asycipc/demo";
  }
  if (const char* xdg = std::getenv("XDG_RUNTIME_DIR")) {
    if (*xdg) {
      return std::string(xdg) + "/asycipc/demo";
    }
  }
  return "/run/user/" + std::to_string(geteuid()) + "/asycipc/demo";
#endif
}

void SendFrames(PipeHandle& handle, const std::vector<Frame>& frames,
                std::mutex& write_mutex);
void SendMaintenance(Channel& channel, PipeHandle& handle,
                     std::mutex& write_mutex);

constexpr std::size_t kFrameHeaderSize = 24;
constexpr std::size_t kFrameTrailerSize = 4;

std::size_t NextFrameSize(const std::vector<uint8_t>& buffer) {
  if (buffer.size() < kFrameHeaderSize) {
    return 0;
  }
  const uint16_t magic = static_cast<uint16_t>(static_cast<uint16_t>(buffer[0]) << 8) |
                         static_cast<uint16_t>(buffer[1]);
  if (magic != kFrameMagic) {
    throw std::runtime_error("invalid frame magic in stream");
  }
  if (buffer[2] != kProtocolVersion) {
    throw std::runtime_error("unsupported frame version in stream");
  }
  const uint32_t payload_length =
      (static_cast<uint32_t>(buffer[14]) << 24) |
      (static_cast<uint32_t>(buffer[15]) << 16) |
      (static_cast<uint32_t>(buffer[16]) << 8) |
      static_cast<uint32_t>(buffer[17]);
  const uint16_t ext_length = static_cast<uint16_t>((static_cast<uint16_t>(buffer[22]) << 8) |
                                                    static_cast<uint16_t>(buffer[23]));

  const std::size_t total = kFrameHeaderSize + static_cast<std::size_t>(ext_length) +
                            static_cast<std::size_t>(payload_length) +
                            kFrameTrailerSize;
  if (buffer.size() < total) {
    return 0;
  }
  return total;
}

bool DrainClientBuffer(Channel& channel, PipeHandle& handle,
                       std::deque<std::vector<uint8_t>>& inbox,
                       std::mutex& write_mutex, std::vector<uint8_t>& buffer) {
  bool processed = false;
  while (true) {
    const std::size_t frame_size = NextFrameSize(buffer);
    if (frame_size == 0 || buffer.size() < frame_size) {
      break;
    }
    std::vector<uint8_t> frame_bytes(buffer.begin(),
                                     buffer.begin() + static_cast<std::ptrdiff_t>(frame_size));
    buffer.erase(buffer.begin(),
                 buffer.begin() + static_cast<std::ptrdiff_t>(frame_size));

    Frame frame = Frame::Deserialize(frame_bytes);
    auto result = channel.HandleInboundFrame(frame);
    SendFrames(handle, result.control_frames, write_mutex);
    if (result.application_frame.has_value()) {
      inbox.push_back(result.application_frame->payload);
    }
    SendMaintenance(channel, handle, write_mutex);
    processed = true;
  }
  return processed;
}

void SendFrames(PipeHandle& handle, const std::vector<Frame>& frames,
                std::mutex& write_mutex) {
  for (const auto& frame : frames) {
    auto bytes = frame.Serialize();
    std::lock_guard<std::mutex> lock(write_mutex);
    handle.Write(bytes);
  }
}

void SendMaintenance(Channel& channel, PipeHandle& handle, std::mutex& write_mutex) {
  auto retries = channel.PollRetries();
  if (!retries.empty()) {
    std::vector<Frame> frames;
    frames.reserve(retries.size());
    for (const auto& retry : retries) {
      frames.push_back(retry.frame);
    }
    SendFrames(handle, frames, write_mutex);
  }
  if (channel.ShouldSendHeartbeat()) {
    SendFrames(handle, {Frame::MakeHeartbeat()}, write_mutex);
    channel.MarkHeartbeatSent();
  }
}

bool PumpServer(Channel& channel, PipeHandle& handle,
                std::deque<std::vector<uint8_t>>& inbox,
                std::mutex& write_mutex, std::vector<uint8_t>& read_buffer) {
  while (true) {
    if (DrainClientBuffer(channel, handle, inbox, write_mutex, read_buffer)) {
      return true;
    }

    std::vector<uint8_t> payload;
    try {
      payload = handle.Read();
    } catch (const std::exception& ex) {
      std::cerr << "[client] read failed: " << ex.what() << std::endl;
      return false;
    }
    if (payload.empty()) {
      return false;
    }

    read_buffer.insert(read_buffer.end(), payload.begin(), payload.end());
  }
}

}  // namespace

int main(int argc, char** argv) {
  std::string pipe_endpoint = DefaultEndpoint();
  if (argc > 1) {
    pipe_endpoint = argv[1];
  }

  auto handle = ConnectNamedPipeEndpoint({pipe_endpoint});
  std::cout << "[client] connected to " << pipe_endpoint << std::endl;

  handle->Write(ToBytes("HELLO\n"));
  std::vector<uint8_t> handshake_accum;
  std::vector<uint8_t> initial_buffer;
  while (true) {
    auto chunk = handle->Read();
    if (chunk.empty()) {
      throw std::runtime_error("server closed during handshake");
    }
    handshake_accum.insert(handshake_accum.end(), chunk.begin(), chunk.end());
    auto newline_it = std::find(handshake_accum.begin(), handshake_accum.end(),
                                static_cast<uint8_t>('\n'));
    if (newline_it == handshake_accum.end()) {
      continue;
    }

    std::string ack_text(handshake_accum.begin(), newline_it + 1);
    std::cout << "[client] handshake ack: " << ack_text << std::endl;
    initial_buffer.assign(newline_it + 1, handshake_accum.end());
    break;
  }

  SessionContext session;
  session.SetSessionId("client-session");
  session.TransitionTo(SessionState::kNegotiating);
  session.TransitionTo(SessionState::kActive);

  Channel channel(session, nullptr);
#if !defined(__APPLE__)
  auto window = std::make_shared<SlidingWindow>(256, std::chrono::milliseconds(500), 5);
  auto heartbeat = std::make_shared<HeartbeatMonitor>(std::chrono::milliseconds(5000), 3);
  auto backpressure = std::make_shared<BackpressureController>(256, 128);
  TelemetryCenter telemetry;
  channel.SetReliabilityModules(window, heartbeat, backpressure, &telemetry);
#endif
  channel.SetTimeProvider([] { return std::chrono::steady_clock::now(); });

  std::deque<std::vector<uint8_t>> inbox;
  channel.SetMessageHandler([&](Channel&, const Frame& frame) {
    inbox.push_back(frame.payload);
    return std::vector<Frame>{};
  });
  channel.SetErrorHandler([](const std::string& msg) {
    std::cerr << "[client] error: " << msg << std::endl;
  });

  std::mutex write_mutex;

  std::vector<uint8_t> read_buffer = std::move(initial_buffer);

  auto wait_for_message = [&]() {
    while (inbox.empty()) {
      if (!PumpServer(channel, *handle, inbox, write_mutex, read_buffer)) {
        throw std::runtime_error("connection closed");
      }
    }
    auto data = std::move(inbox.front());
    inbox.pop_front();
    return data;
  };

  auto send_payload = [&](const std::vector<uint8_t>& data) {
    auto frames = channel.MakeDataFrames(data);
    SendFrames(*handle, frames, write_mutex);
    SendMaintenance(channel, *handle, write_mutex);
  };

  try {
    std::cout << "[client] phase 1: respond to server pings" << std::endl;
    for (int i = 0; i < 100; ++i) {
      auto payload = wait_for_message();
      std::string text = ToString(payload);
      std::cout << "  [client] received: " << text << std::endl;
      std::string reply = "client pong " + std::to_string(i);
      send_payload(ToBytes(reply));
    }

    std::cout << "[client] phase 2: initiate pings" << std::endl;
    for (int i = 0; i < 100; ++i) {
      std::string ping = "client ping " + std::to_string(i);
      send_payload(ToBytes(ping));

      auto payload = wait_for_message();
      std::string text(payload.begin(), payload.end());
      std::cout << "  [client] received: " << text << std::endl;
    }

    std::cout << "[client] phase 3: handle 4MB transfers" << std::endl;
    for (int i = 0; i < 100; ++i) {
      auto payload = wait_for_message();
      std::cout << "  [client] chunk " << std::setw(3) << i
                << " size=" << payload.size() << std::endl;
      auto inverted = payload;
      for (auto& byte : inverted) {
        byte = static_cast<uint8_t>(~byte);
      }
      send_payload(inverted);
    }
  } catch (const std::exception& ex) {
    std::cerr << "[client] termination: " << ex.what() << std::endl;
  }

  handle.reset();

  std::cout << "[client] demo completed" << std::endl;
  return 0;
}
